30 实践课-智能体异常处理机制设计

智能体异常处理机制设计

关联：索引

术语小抄（初学者版）

【教师参考】异常注入清单（用于演示复现→定位→修复→回归）

| :-- | :-- | :-- | :-- |
| 查询工具超时 | 工具返回 ok=false + error.code=TIMEOUT | 触发重试；到达上限后降级/提示 | attempt/max_attempts、meta.retry_decision、trace_id |
| 控制工具不可用 | 控制工具返回 SERVICE_UNAVAILABLE 或回执缺失/不匹配 | 优先级 1：拒绝/停止/降级为“只给建议” | 明确“未下发控制”+ 错误码 + trace_id |
| 权限不足 | 工具返回 PERMISSION_DENIED | 不重试；直接拒绝并提示申请权限 | retry_decision=no_retry_non_retryable |
| 入参缺失/类型错 | 工具返回 INPUT_INVALID | 不重试；提示补齐槽位/重新输入 | error.code=INPUT_INVALID + 提示内容 |
| 意图模糊 | 构造“前两名分数接近/最高分偏低”的 scores | 触发澄清提示（候选≤3） | need_clarify=true、候选意图列表、parse_trace_id |
| LLM 不可用 | LLM 调用抛异常/返回不可用状态（模拟断网/额度） | 走离线路由与可解释拒绝 | 输出包含 reason 与下一步选项 |

课程思政融入点（口径统一）：

类别 典型表现 最小证据 推荐处理
A. 工具执行失败（控制类） 机械臂控制工具返回失败/超时/回执对不上 tool_name + trace_id + error.code + 回执字段 先停/拒绝，再考虑重试；必要时降级为“只给操作建议、不下发控制”
B. 工具执行失败（查询类） 查询规则/知识库超时或无结果 trace_id + error.code 或 hits=0 可重试（瞬态），或降级到离线规则库/提示人工确认
C. 意图识别模糊 低置信度、多候选意图接近 parse_trace_id + 候选列表 先澄清（提示功能），避免误执行
D. LLM 不可用 调用接口失败/额度/网断 错误码/异常栈 + parse_trace_id 降级：离线指令库 + 简化决策 + 可解释拒绝
  1. 控制类工具失败（安全优先）：先阻断误控风险（拒绝/停止/降级），再考虑恢复。
  2. LLM 不可用（系统可用性）：保证“能给出稳定可控的答复”，必要时只做离线决策或解释。
  3. 意图识别模糊（正确性）：通过澄清提高正确率，宁可多问一句也不误执行。
  4. 查询类工具失败（体验）：可重试与降级并用，确保给出可行动的替代方案。

自检要点：

  1. 只重试瞬态失败，不重试永久失败：例如网络超时可重试，参数缺失/权限不足不可重试。
  2. 次数有限且可配置：默认 2–3 次足够；控制类工具更保守（甚至 0 次），查询类可略多。
  3. 间隔策略要解释得通：固定间隔（易理解）或指数退避（更抗抖动），必要时加抖动避免雪崩。
  4. 每次失败都要留下证据：失败原因、次数、等待时间、trace_id/parse_trace_id 必须可追踪，便于回归。

触发条件（建议落地为“错误码/异常类型白名单”）：

目标：把“工具调用”统一成一个入口 call_tool_with_retry，并把重试策略显式化（次数/间隔/触发条件）。

建议放置位置（不强制）：04_sorting_agent_practice/resilience.py，由 app_with_intent.py 或工具路由层调用。

import random
import time
from dataclasses import dataclass
from typing import Any, Callable, Dict, Iterable, Optional, Tuple

@dataclass(frozen=True)
class RetryPolicy:
    max_attempts: int
    base_delay_s: float
    backoff: str  # "fixed" | "exponential"
    jitter_s: float
    retryable_error_codes: Tuple[str, ...]

def _compute_delay_s(policy: RetryPolicy, attempt_index: int) -> float:
    if policy.backoff == "fixed":
        raw = policy.base_delay_s
    elif policy.backoff == "exponential":
        raw = policy.base_delay_s * (2 ** attempt_index)
    else:
        raise ValueError(f"unsupported backoff: {policy.backoff}")
    return max(0.0, raw + random.uniform(0.0, policy.jitter_s))

def _is_retryable(tool_result: Dict[str, Any], retryable_error_codes: Iterable[str]) -> bool:
    if tool_result.get("ok") is True:
        return False
    err = tool_result.get("error") or {}
    code = err.get("code")
    return isinstance(code, str) and code in set(retryable_error_codes)

def call_tool_with_retry(
    tool_name: str,
    tool_fn: Callable[ [Dict[str, Any]], Dict[str, Any] ],
    payload: Dict[str, Any],
    *,
    policy: RetryPolicy,
    trace_id_seed: str,
    extra: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    extra = extra or {}
    last_out: Dict[str, Any] = {}

    for attempt in range(policy.max_attempts):
        out = tool_fn(payload)
        out = dict(out)
        out.setdefault("tool_name", tool_name)
        out.setdefault("attempt", attempt + 1)
        out.setdefault("max_attempts", policy.max_attempts)
        out.setdefault("trace_id", out.get("trace_id") or f"{trace_id_seed}-{attempt+1}")
        out.setdefault("meta", {})
        out["meta"] = dict(out["meta"])
        out["meta"].update(extra)

        last_out = out
        if out.get("ok") is True:
            return out

        if not _is_retryable(out, policy.retryable_error_codes):
            out.setdefault("meta", {})
            out["meta"] = dict(out["meta"])
            out["meta"]["retry_decision"] = "no_retry_non_retryable"
            return out

        if attempt >= policy.max_attempts - 1:
            out.setdefault("meta", {})
            out["meta"] = dict(out["meta"])
            out["meta"]["retry_decision"] = "no_retry_reach_max_attempts"
            return out

        delay_s = _compute_delay_s(policy, attempt_index=attempt)
        out.setdefault("meta", {})
        out["meta"] = dict(out["meta"])
        out["meta"]["retry_decision"] = "retry"
        out["meta"]["retry_sleep_s"] = round(delay_s, 3)
        time.sleep(delay_s)

    return last_out

逐段解释与自检要点：

可复制的最小演示（用假工具模拟失败→重试→成功）：

import random
from typing import Any, Dict

from resilience import RetryPolicy, call_tool_with_retry

def flaky_query_tool(payload: Dict[str, Any]) -> Dict[str, Any]:
    if random.random() < 0.6:
        return {"ok": False, "error": {"code": "TIMEOUT", "message": "query timeout"}}
    return {"ok": True, "data": {"rule_id": "R-GLASS-01", "payload": payload}}

policy = RetryPolicy(
    max_attempts=3,
    base_delay_s=0.2,
    backoff="exponential",
    jitter_s=0.1,
    retryable_error_codes=("TIMEOUT", "SERVICE_UNAVAILABLE", "RATE_LIMIT", "CONNECTION_ERROR"),
)

out = call_tool_with_retry(
    tool_name="query_rule",
    tool_fn=flaky_query_tool,
    payload={"item_type": "glass"},
    policy=policy,
    trace_id_seed="demo-trace",
    extra={"parse_trace_id": "demo-parse"},
)
print(out)

说明与自检要点：

运行（Windows PowerShell，示例）：

cd ".\04_sorting_agent_practice"
python .\demo_retry.py

命令解释与自检要点：

1）为你们组的一个“查询类工具”设计重试策略：写出 RetryPolicy 的参数选择理由，并列出 2 个可重试与 2 个不可重试的错误码。
2）为你们组的一个“控制类工具”设计“不重试/谨慎重试”策略：说明为什么不能盲目重试，并写出“必须阻断误控”的 2 条规则（例如回执不匹配立即停止）。

  1. 把“意图不确定”变成可交互的澄清提示（引导补充 + 候选意图）。
  2. 把“LLM 不可用”变成可控的降级路径（离线指令库 + 简化决策 + 可解释拒绝）。
  3. 用 AI 协同生成异常处理代码，并用标注的异常场景数据把机制补齐与回归。

澄清提示的三条原则：

  1. 先给候选意图，再问补充信息：减少用户思考成本。
  2. 候选不超过 3 个：超过 3 个就退回“让用户改写/补充关键字段”。
  3. 每次澄清都要可落地到槽位：例如要 task_id、要 item_type、要 “设备/工位”。

最小输出口径（建议直接写进路由层）：

最小可运行：基于“置信度/差值阈值”的澄清判定（示例代码，可复制）

from dataclasses import dataclass
from typing import Dict, List, Tuple

@dataclass(frozen=True)
class IntentScore:
    intent: str
    score: float

def should_clarify(scores: List[IntentScore], *, min_top_score: float, min_margin: float) -> Tuple[bool, Dict]:
    if not scores:
        return True, {"reason": "no_candidates"}

    scores_sorted = sorted(scores, key=lambda x: x.score, reverse=True)
    top = scores_sorted[0]
    second = scores_sorted[1] if len(scores_sorted) > 1 else IntentScore(intent="__none__", score=0.0)
    margin = top.score - second.score

    if top.score < min_top_score:
        return True, {"reason": "top_score_low", "top_score": top.score}
    if margin < min_margin:
        return True, {"reason": "margin_low", "top_score": top.score, "second_score": second.score, "margin": margin}
    return False, {"reason": "confident", "top_intent": top.intent, "top_score": top.score, "margin": margin}

逐段解释与自检要点：

澄清提示模板（分拣场景示例，可直接用在输出）

我不确定你要做哪一种操作（为避免误执行，需要你确认一下）：
1）查询分拣规则（例：查一下 glass 的分拣规则）
2）提交异常反馈（例：提交反馈 t001 玻璃破损）
3）设备控制（例：让机械臂回零/夹取/移动到工位 A）

请回复序号（1/2/3），或补充关键字段：物品类型(item_type) / 任务号(task_id) / 设备动作(action)。

自检要点：

离线指令库（示例：关键词→意图/动作/槽位），只用标准库即可运行：

import re
from dataclasses import dataclass
from typing import Dict, Optional

@dataclass(frozen=True)
class OfflineDecision:
    ok: bool
    intent: str
    slots: Dict[str, str]
    reason: str

def offline_route(text: str) -> OfflineDecision:
    t = text.strip().lower()
    if not t:
        return OfflineDecision(ok=False, intent="UNKNOWN", slots={}, reason="empty_text")

    m = re.search(r"\b(t\d{3,})\b", t)
    if ("提交" in text or "反馈" in text) and m:
        return OfflineDecision(ok=True, intent="SUBMIT_FEEDBACK", slots={"task_id": m.group(1)}, reason="keyword_feedback+task_id")

    m2 = re.search(r"\b(glass|metal|plastic|paper)\b", t)
    if ("规则" in text or "怎么分拣" in text) and m2:
        return OfflineDecision(ok=True, intent="QUERY_RULE", slots={"item_type": m2.group(1)}, reason="keyword_rule+item_type")

    if any(k in text for k in ["回零", "急停", "停止", "复位", "移动", "夹取"]):
        return OfflineDecision(ok=False, intent="CONTROL_DEVICE", slots={}, reason="control_requires_llm_or_confirmation")

    return OfflineDecision(ok=False, intent="GENERAL", slots={}, reason="no_match_fallback_to_explain")

逐段解释与自检要点：

运行（Windows PowerShell，示例）：

python -c "from offline_fallback import offline_route; print(offline_route('查一下 glass 的分拣规则')); print(offline_route('提交反馈 t001 玻璃破损')); print(offline_route('让机械臂回零'))"

命令解释与自检要点：

可解释拒绝模板（LLM 不可用/控制风险）

当前大模型不可用/不稳定，为保证安全我不会直接下发设备控制指令。
你可以选择：
1）改为查询规则/提交反馈（我可以离线处理）
2）补充明确指令并确认（例如：动作=回零，设备=arm_01，是否确认=是）
3）转为人工操作：请按现场 SOP 执行急停/复位流程
  1. 让 AI 先给“机制模板”（场景→规则→优先级→证据字段）。
  2. 让 AI 生成可运行代码（重试/澄清/降级），要求只用标准库或项目已用依赖。
  3. 人工审计四件事：安全边界、不可重试边界、证据字段、是否破坏权限/控制链路。
  4. 用标注异常场景数据回归：同一批样本跑前后对比（覆盖率、误控风险是否下降）。
你是工业分拣智能体的异常处理机制设计助手。请输出：
1）异常场景分类表（至少 6 类），并给出处理优先级（控制失败最高）。
2）每类异常的处理规则：触发条件（错误码/阈值/特征）、动作（重试/澄清/降级/拒绝）、输出证据字段（trace_id/parse_trace_id/error.code）。
3）给出 Python 3.10 标准库实现的核心代码：
   - call_tool_with_retry（重试次数/间隔/触发条件可配置）
   - should_clarify（低置信度与小差值触发澄清）
   - offline_route（LLM 不可用时的离线路由）
要求：代码可直接复制运行；不得省略关键错误码与安全拒绝逻辑；给出最小 demo。

最小标注格式（JSON Lines，示例，可直接复制生成小样）

{"text":"让机械臂夹取 glass 并放到 2 号箱","case":"CONTROL_TOOL_FAIL","llm_available":true,"tool_result":{"ok":false,"error":{"code":"SERVICE_UNAVAILABLE","message":"arm service down"}},"expected":"SAFE_REJECT_OR_DEGRADE"}
{"text":"查一下 glass 的分拣规则","case":"QUERY_TOOL_TIMEOUT","llm_available":true,"tool_result":{"ok":false,"error":{"code":"TIMEOUT","message":"db timeout"}},"expected":"RETRY_THEN_DEGRADE"}
{"text":"帮我处理一下","case":"INTENT_AMBIGUOUS","llm_available":true,"intent_scores":[["QUERY_RULE",0.41],["SUBMIT_FEEDBACK",0.39],["GENERAL",0.20]],"expected":"CLARIFY_WITH_CANDIDATES"}
{"text":"查一下 metal 的规则","case":"LLM_DOWN","llm_available":false,"expected":"OFFLINE_ROUTE"}

每组现场演示（最低要求）：

  1. 梳理自选题场景的异常场景清单（至少 3 类），设计对应的处理机制（含处理优先级）。

  2. 使用 AI 大模型生成异常处理代码（重试、提示、降级），并完成人工审计与最小复验。

  3. 接入标注的异常场景数据（5–10 条即可），用数据完善处理逻辑（阈值/错误码白名单/拒绝规则）。

  4. 记录机制设计思路（为什么这样设计、风险是什么、怎么回归验证）。

  5. 生成智能体异常处理机制设计模板（含场景、规则、处理逻辑、优先级与证据字段）。

  6. 生成异常处理核心代码（重试机制、提示功能、降级方案），要求可运行并符合安全边界。

  7. 针对标注的异常场景数据，给出机制完善建议（阈值、规则补丁、回归用例）。

  8. 解答异常处理代码开发中的问题（定位“该重试/不该重试”、如何输出可解释证据）。

作业（布置）

1）提交自选题场景的异常处理机制设计文档（含场景分类、处理规则、优先级）。
2）提交异常处理代码（AI 生成 + 人工优化版）、各异常场景测试截图。
3）提交标注异常场景数据的使用记录，说明如何基于数据完善处理机制。

Markdown 与代码自检清单（提交前必须过一遍）